package com.bky.mq.base;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.MessageListener;

public abstract class BaseConsumer<T> implements MessageListener<String, String> {
    protected final Logger LOG = LoggerFactory.getLogger("BaseConsumer");
    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord) {
        try {
            if (null == consumerRecord){
                return;
            }
            if (null == consumerRecord.value()){
                return;
            }
            JSONObject jsonObject = JSONObject.parseObject(consumerRecord.value());
            if (null == jsonObject){
                return;
            }
            T t = checkBusinessData(jsonObject);
            if (null != t){
                processBusiness(t);
            }
        } catch (Exception e) {
            LOG.error(getExceptionMessage(), e);
        }
    }

    /**
     * 设置异常提示语
     *
     * @return
     */
    protected abstract String getExceptionMessage();

    /**
     * 检查业务数据是否合法
     *
     * @param jsonObject
     * @return
     */
    protected abstract T checkBusinessData(JSONObject jsonObject);

    /**
     * 处理业务流程
     */
    protected abstract void processBusiness(T t);
}
